#include "kwdb/ilp_client.h"
#include "security_util.h"
#include <sstream>
#include <iomanip>
#include <chrono>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <cstring>
#include <iostream>
#include <zlib.h>

namespace kwdb {

// Helper function for GZIP compression
std::string gzip_compress(const std::string& data) {
    z_stream zs;
    memset(&zs, 0, sizeof(zs));

    // Initialize for gzip format (16 + MAX_WBITS)
    if (deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
                     16 + MAX_WBITS, 8, Z_DEFAULT_STRATEGY) != Z_OK) {
        return data; // Return uncompressed on error
    }

    zs.next_in = (Bytef*)data.data();
    zs.avail_in = data.size();

    int ret;
    char outbuffer[32768];
    std::string compressed;

    do {
        zs.next_out = reinterpret_cast<Bytef*>(outbuffer);
        zs.avail_out = sizeof(outbuffer);

        ret = deflate(&zs, Z_FINISH);

        if (compressed.size() < zs.total_out) {
            compressed.append(outbuffer, zs.total_out - compressed.size());
        }
    } while (ret == Z_OK);

    deflateEnd(&zs);

    if (ret != Z_STREAM_END) {
        return data; // Return uncompressed on error
    }

    return compressed;
}

// Implementation class
class IlpClient::IlpClientImpl {
public:
    IngestConfig config_;
    std::string host_;
    int port_;

    IlpClientImpl(const IngestConfig& config) : config_(config) {
        // Parse endpoint (host:port)
        std::string endpoint = config.ingest_endpoint;
        size_t colon_pos = endpoint.rfind(':');

        if (colon_pos != std::string::npos) {
            host_ = endpoint.substr(0, colon_pos);
            port_ = std::stoi(endpoint.substr(colon_pos + 1));
        } else {
            host_ = endpoint;
            port_ = 9091; // Default ILP port
        }
    }

    Status send_http_request(const std::string& body) {
        // Create socket
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (sockfd < 0) {
            return Status(-1, "Failed to create socket");
        }

        // Resolve hostname
        struct hostent* server = gethostbyname(host_.c_str());
        if (server == nullptr) {
            close(sockfd);
            return Status(-1, "Failed to resolve host: " + host_);
        }

        // Connect to server
        struct sockaddr_in serv_addr;
        memset(&serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        memcpy(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
        serv_addr.sin_port = htons(port_);

        if (connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
            close(sockfd);
            return Status(-1, "Failed to connect to " + host_ + ":" + std::to_string(port_));
        }

        // Set socket timeout
        struct timeval tv;
        tv.tv_sec = config_.timeout_ms / 1000;
        tv.tv_usec = (config_.timeout_ms % 1000) * 1000;
        setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
        setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));

        // Prepare body (with optional GZIP compression)
        std::string final_body = body;
        bool is_compressed = false;

        if (config_.security.gzip && body.size() > 100) { // Only compress if worth it
            std::string compressed = gzip_compress(body);
            if (compressed.size() < body.size()) {
                final_body = compressed;
                is_compressed = true;
            }
        }

        // Build HTTP request
        std::stringstream request;
        request << "POST /write HTTP/1.1\r\n";
        request << "Host: " << host_ << ":" << port_ << "\r\n";
        request << "Content-Type: text/plain; charset=utf-8\r\n";
        request << "Content-Length: " << final_body.size() << "\r\n";

        // Add compression header if needed
        if (is_compressed) {
            request << "Content-Encoding: gzip\r\n";
        }

        // Add authentication header if configured
        if (config_.security.has_auth()) {
            request << "Authorization: " << config_.security.get_auth_header() << "\r\n";
        }

        request << "User-Agent: KWDB-C++-SDK/0.4.0\r\n";
        request << "Connection: close\r\n";
        request << "\r\n";
        request << final_body;

        std::string request_str = request.str();

        // Send request
        ssize_t bytes_sent = send(sockfd, request_str.c_str(), request_str.size(), 0);
        if (bytes_sent < 0 || (size_t)bytes_sent != request_str.size()) {
            close(sockfd);
            return Status(-1, "Failed to send request");
        }

        // Read response
        char buffer[4096];
        std::string response;
        ssize_t bytes_received;

        while ((bytes_received = recv(sockfd, buffer, sizeof(buffer) - 1, 0)) > 0) {
            buffer[bytes_received] = '\0';
            response.append(buffer);

            // Check if we have complete headers
            if (response.find("\r\n\r\n") != std::string::npos) {
                break;
            }
        }

        close(sockfd);

        // Parse HTTP status code
        if (response.empty()) {
            return Status(-1, "Empty response from server");
        }

        // Find status code
        size_t status_start = response.find("HTTP/1.") + 9;
        if (status_start == std::string::npos + 9) {
            return Status(-1, "Invalid HTTP response");
        }

        int status_code = std::stoi(response.substr(status_start, 3));

        if (status_code == 204) {
            return Status(0, "OK");
        } else if (status_code >= 200 && status_code < 300) {
            return Status(0, "OK");
        } else if (status_code >= 400 && status_code < 500) {
            return Status(-1, "Client error: HTTP " + std::to_string(status_code));
        } else {
            return Status(-1, "Server error: HTTP " + std::to_string(status_code));
        }
    }
};

// Public API implementation

IlpClient::IlpClient(const IngestConfig& config)
    : impl_(std::make_unique<IlpClientImpl>(config)) {
    std::cout << "[ILP Client] Initialized for " << config.ingest_endpoint << std::endl;
    if (config.security.gzip) {
        std::cout << "[ILP Client] GZIP compression enabled" << std::endl;
    }
    if (config.security.has_auth()) {
        std::cout << "[ILP Client] Authentication configured" << std::endl;
    }
}

IlpClient::~IlpClient() = default;

Status IlpClient::write(const Point& point) {
    std::string line_protocol = point_to_ilp(point);
    return write_raw(line_protocol);
}

Status IlpClient::batch_write(const std::vector<Point>& points) {
    std::string line_protocol = points_to_ilp(points);
    return write_raw(line_protocol);
}

Status IlpClient::write_raw(const std::string& line_protocol) {
    if (line_protocol.empty()) {
        return Status(-1, "Empty line protocol");
    }

    return impl_->send_http_request(line_protocol);
}

Status IlpClient::flush() {
    // For future batching support
    return Status(0, "OK");
}

// ILP format conversion functions

std::string ilp_escape(const std::string& value, const std::string& context) {
    std::string escaped;
    escaped.reserve(value.size() * 2);

    for (char c : value) {
        if (context == "measurement" || context == "tag_key") {
            // Escape comma, space, equals
            if (c == ',' || c == ' ' || c == '=') {
                escaped += '\\';
            }
        } else if (context == "tag_value") {
            // Escape comma, space, equals
            if (c == ',' || c == ' ' || c == '=') {
                escaped += '\\';
            }
        } else if (context == "field_key") {
            // Escape comma, space, equals
            if (c == ',' || c == ' ' || c == '=') {
                escaped += '\\';
            }
        } else if (context == "field_value") {
            // String field values: escape double quotes and backslashes
            if (c == '"' || c == '\\') {
                escaped += '\\';
            }
        }
        escaped += c;
    }

    return escaped;
}

std::string point_to_ilp(const Point& point) {
    std::stringstream ss;

    // Measurement
    ss << ilp_escape(point.get_measurement(), "measurement");

    // Tags (sorted for consistency)
    const auto& tags = point.get_tags();
    if (!tags.empty()) {
        for (const auto& [key, value] : tags) {
            ss << "," << ilp_escape(key, "tag_key")
               << "=" << ilp_escape(value, "tag_value");
        }
    }

    // Space separator
    ss << " ";

    // Fields
    const auto& fields = point.get_fields();
    bool first = true;
    for (const auto& [key, value] : fields) {
        if (!first) ss << ",";
        ss << ilp_escape(key, "field_key") << "=";

        // Format field value
        // For simplicity, we're using double. In production, support int64 with 'i' suffix
        ss << std::fixed << std::setprecision(6) << value;

        first = false;
    }

    // Timestamp (nanoseconds)
    auto timestamp_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
        point.get_timestamp().time_since_epoch()
    ).count();

    ss << " " << timestamp_ns;

    return ss.str();
}

std::string points_to_ilp(const std::vector<Point>& points) {
    std::stringstream ss;

    for (const auto& point : points) {
        ss << point_to_ilp(point) << "\n";
    }

    return ss.str();
}

} // namespace kwdb